#ifndef STREAM_BUS_HPP
#define STREAM_BUS_HPP

// ======================= 头文件依赖 =======================
// 这里只依赖标准库 + 你现有工程里的 PacketBlob / VideoPacketReader 以及第三方：
// - moodycamel::BlockingConcurrentQueue（阻塞并发队列）
// - nlohmann::json（JSON 序列化）
// 注意：本文件不直接包含 FFmpeg 头文件；AVPacket 类型由 VideoPacketReader.hpp 引入。
#include <atomic>
#include <cstdint>
#include <cstdio>
#include <cstring>
#include <limits>
#include <memory>
#include <string>
#include <thread>
#include <utility>
#include <vector>
#include <chrono>
#include <iostream>

#include "PacketBlob.hpp"                               // 你项目里的“承载一帧”的轻量结构（元数据 + extradata + data 的智能指针包装）
#include "VideoPacketReader.hpp"                        // 你项目里的拉流器，内部封装 FFmpeg 的打开/读取/回调
#include "../concurrentqueue/blockingconcurrentqueue.h" // moodycamel 阻塞并发队列
#include "../config/nlohmannjson.hpp"                   // 你项目内对 nlohmann::json 的包装头

/**
 * @class StreamBus
 * @brief
 *   “视频包总线”的**单例**：统一负责**从 VideoPacketReader 拉包 → 在回调中深拷贝到 PacketBlob → 指针入队**。
 *
 * 主要职责/数据流：
 *   1) Open(url)：创建并打开 VideoPacketReader，安装 packet/error 回调；
 *   2) Start()：启动一个后台拉流线程，在循环中反复调用 reader_->ReadNextPacket()；
 *      - 读到包后，VideoPacketReader 内部触发我们安装的 packet 回调；
 *      - 回调里**只做轻量操作**：拷元数据、按需深拷贝 extradata 和 data，最后把 `shared_ptr<PacketBlob>` 入队；
 *   3) 使用方在另一个线程中通过 WaitDequeue/TryDequeue 取出指针，随后可调用 BuildFrameChunkBuffer
 *      将其序列化为：Header(4B 小端 JSON长度) + JSON + extradata + data，然后写 socket/文件即可。
 *
 * 线程与内存模型：
 *   - 队列元素类型是 `std::shared_ptr<PacketBlob>`，入队/出队只搬运指针，**零拷贝共享结构体**；
 *   - `opened_` / `running_` / `stop_` / 统计计数使用 `std::atomic` 保证跨线程可见性；
 *   - 拉流线程只负责调用 `ReadNextPacket()`，**繁重/阻塞**工作（网络发送/磁盘写入）不应放在回调里。
 *
 * 关键行为（按照你的要求）：
 *   - 删除“只在首包带 extradata”的策略；改为：**如果信息里有 extradata，则每个包都携带**。
 *   - `opened_`/`running_` 使用原子类型，并在访问时使用 `load/store`（或等价的原子赋值）。
 */
class StreamBus
{
public:
    //==================== 单例获取 ====================

    /**
     * @brief 获取全局唯一实例（Meyers' Singleton）
     * C++11 之后局部静态变量的初始化是线程安全的。
     */
    static StreamBus &Instance()
    {
        static StreamBus inst;
        return inst;
    }

    //==================== FFmpeg 全局初始化（可选） ====================

    /**
     * @brief 显式初始化 FFmpeg 网络子系统（可多次调用，内部幂等）
     * 说明：VideoPacketReader::GlobalInit() 内部已做静态保护。
     * 场景：你希望在程序启动处显式初始化，便于日志/顺序控制。
     */
    void InitFFmpegNetwork()
    {
        VideoPacketReader::GlobalInit();
    }

    /**
     * @brief 显式反初始化 FFmpeg 网络子系统（可多次调用，内部幂等）
     */
    void DeinitFFmpegNetwork()
    {
        VideoPacketReader::GlobalCleanup();
    }

    //==================== 打开 / 启动 / 停止 / 关闭 ====================

    /**
     * @brief 打开一个输入源并安装回调，但不启动读线程
     *
     * @param url        输入源（文件/RTSP/RTMP/HTTP(S)/UDP 等）
     * @param net        网络/探测参数（VideoPacketReader::NetworkOptions）
     * @param source_id  源标识，会写入 JSON，方便下游区分路由/调试
     * @return true 打开成功；false 打开失败（会清理 reader_）
     *
     * 调用流程：
     *   - 若之前已打开，先 Stop(true) + Close()，确保无悬挂线程/资源；
     *   - 创建新的 VideoPacketReader 实例，记录 source_id，重置帧序号；
     *   - 安装 **packet 回调**（深拷贝 + 入队，**每包**按需附带 extradata）；
     *   - 安装 **error 回调**（打印即可，避免阻塞）；
     *   - 调用 reader_->Open(url, net) 真正打开输入；
     *   - 成功后把 opened_ 置 true（原子写）。
     *
     * 重要：本函数不启动线程；你可以在成功后调用 Start()。
     */
    bool Open(const std::string &url,
              const media_types::NetworkOptions &net = media_types::NetworkOptions(),
              const std::string &source_id = "default")
    {
        // 1) 若之前处于打开/运行状态，先停止并关闭，确保干净
        Stop(/*drain=*/true);
        Close();

        // 2) 创建“拉流器”实例（unique_ptr 所有权，确保异常安全）
        reader_ = std::make_unique<VideoPacketReader>(); // ← 这里会在构造中完成 FFmpeg 基础准备
        source_id_ = source_id;                          // ← 记下本源标识，写进每个 PacketBlob 的 JSON
        seq_.store(0, std::memory_order_release);        // ← 自增帧序号从 0 开始（原子写，供多线程读）

        // 3) 安装“收到包时”的回调：**只做轻量拷贝 + 入队**
        //    回调发生在拉流线程上下文中，不要做耗时/阻塞操作（网络发送/磁盘 IO）。
        reader_->SetPacketCallback([this](const AVPacket *pkt,
                                          const media_types::StreamInfo &info)
                                   {
            // 3.1 构造一个 PacketBlob 智能指针承载本帧
            auto b = std::make_shared<PacketBlob>();

            // 3.2 填充“来源与序列号”
            b->source_id = source_id_;                    // 源标识（用于 JSON 里）
            b->seq       = seq_.fetch_add(1, std::memory_order_acq_rel); // 原子自增，保证跨线程有序递增

            // 3.3 拷贝“编解码参数”（注意：这里是按值复制字符串与整数，成本很低）
            b->codec_id        = static_cast<int32_t>(info.codec_id);
            b->codec_name      = info.codec_name;
            b->codec_long_name = info.codec_long_name;
            b->width           = info.width;
            b->height          = info.height;
            b->fps             = info.fps;
            b->bit_rate        = info.bit_rate;
            b->pix_fmt_name    = info.pix_fmt_name;
            b->SetTimeBase(info.time_base.num, info.time_base.den); // 时间基（num/den）

            // 3.4 **按你的新策略**：每个包如果 info.extradata 非空，就深拷贝一份到 PacketBlob
            //     extradata 常用于 H.264/H.265 的 SPS/PPS/VPS；每帧都带，接收端随时可初始化/切流。
            if (!info.extradata.empty()) {
                b->DeepCopyExtra(info.extradata.data(),
                                 static_cast<int32_t>(info.extradata.size()));
            }

            // 3.5 填充“包级时间戳/持续时长/关键帧标记”，并深拷贝压缩数据
            if (pkt) {
                b->pts         = (pkt->pts == AV_NOPTS_VALUE) ? PacketBlob::kNoTimestamp : pkt->pts;
                b->dts         = (pkt->dts == AV_NOPTS_VALUE) ? PacketBlob::kNoTimestamp : pkt->dts;
                b->duration    = pkt->duration;
                b->pos         = pkt->pos;
                b->is_keyframe = (pkt->flags & AV_PKT_FLAG_KEY) != 0;

                // 深拷贝压缩帧数据（可能为 Annex-B；VideoPacketReader 可已统一过）
                if (pkt->data && pkt->size > 0) {
                    b->DeepCopyData(pkt->data, pkt->size);
                }
            }

            // 3.6 入队：采用“软上限（满了丢最旧）”策略，追最新，避免积压
            Enqueue(std::move(b)); });

        // 4) 安装“错误回调”：只记录日志即可
        reader_->SetErrorCallback([](int err, const std::string &msg)
                                  { std::cerr << "[StreamBus] reader error(" << err << "): " << msg << "\n"; });

        // 5) 尝试打开源（同步操作，可能有网络/探测耗时；封装在 VideoPacketReader 内部）
        if (!reader_->Open(url, net))
        {
            reader_.reset(); // 打开失败，清理资源
            return false;
        }

        // 6) 标记“已打开”
        opened_.store(true, std::memory_order_release);
        return true;
    }

    /**
     * @brief 启动后台拉流线程（幂等；已启动则直接返回）
     *
     * 线程函数逻辑：
     *   - 在 stop_ 未被置位前，一直调用 reader_->ReadNextPacket()；
     *   - ReadNextPacket() 失败（EOF/错误/被 RequestStop）即退出循环；
     *   - 退出时把 running_ 置 false。
     */
    void Start()
    {
        // 只有“打开成功 + 未在运行”才启动
        if (!opened_.load(std::memory_order_acquire) ||
            running_.load(std::memory_order_acquire))
        {
            return;
        }

        stop_.store(false, std::memory_order_release);
        running_.store(true, std::memory_order_release);

        // 后台线程：只负责“读”。真正的“拷贝+入队”在 packet 回调里做，保证读环尽可能轻。
        worker_ = std::thread([this]()
                              {
            std::cout << "[StreamBus] 拉流线程启动\n";
            while (!stop_.load(std::memory_order_acquire)) {
                if (!reader_ || !reader_->IsOpened()) break;
                // 读一帧；失败（EOF/错误/被外部 RequestStop）则退出
                if (!reader_->ReadNextPacket()) break;
            }
            std::cout << "[StreamBus] 拉流线程退出\n";
            running_.store(false, std::memory_order_release); });
    }

    /**
     * @brief 请求停止拉流并等待线程退出
     *
     * @param drain 是否保留已入队数据：
     *              - true  ：仅停止拉流线程，队列中已累积的数据**保留**（默认）
     *              - false ：在停止后**清空**队列（调用 DrainQueue）
     *
     * 说明：
     *   - 若未在运行（running_ = false），当前实现直接返回；保持与你原先语义一致。
     *   - 想要“即使没在跑也能清队列”，可再调整这里的逻辑（这不在本次你提出的两点更改范围内）。
     */
    void Stop(bool drain = true)
    {
        // 发停止信号 + 打断阻塞IO（即便 running_=false 也发，防止边缘状态）
        stop_.store(true, std::memory_order_release);
        if (reader_)
            reader_->RequestStop();

        // 永远尝试 join（线程可能已退出，但依旧 joinable）
        if (worker_.joinable())
        {
            worker_.join();
        }

        // 标记
        running_.store(false, std::memory_order_release);

        if (!drain)
        {
            DrainQueue();
        }
    }

    /**
     * @brief 关闭当前源并释放资源（可在 Stop 后调用；幂等）
     *
     * 行为：
     *   - 打印统计（由 VideoPacketReader 维护）；
     *   - 关闭 FFmpeg 相关资源；
     *   - 置 opened_ = false。
     */
    void Close()
    {
        if (reader_)
        {
            reader_->PrintStatistics();
            reader_->Close();
            reader_.reset();
        }
        opened_.store(false, std::memory_order_release);
        // 说明：根据你的新要求，已移除 sent_extradata_once_，无需复位。
    }

    //==================== 队列接口（生产者：回调；消费者：你的发送线程） ====================

    /**
     * @brief 设置队列“软上限”容量（>0 有效；达到上限时入队会先丢弃一个最旧元素）
     * 默认值见构造函数。
     */
    void SetCapacitySoft(size_t cap)
    {
        if (cap > 0)
        {
            capacity_soft_ = cap;
        }
    }

    /**
     * @brief 近似队列长度（非严格，供监控/日志观察用）
     */
    size_t SizeApprox() const
    {
        return q_.size_approx();
    }

    /**
     * @brief 阻塞出队（timeout_ms<0 表示无限期等待）
     *
     * @param out        出队得到的 shared_ptr<PacketBlob>
     * @param timeout_ms 超时毫秒；<0 表示无限等待；=0 等价于 TryDequeue
     * @return true 取到；false 超时（仅限 timeout_ms>=0 的情况会返回 false）
     */
    bool WaitDequeue(std::shared_ptr<PacketBlob> &out, int timeout_ms = -1)
    {
        if (timeout_ms < 0)
        {
            q_.wait_dequeue(out); // 该重载无返回值：一定成功
            return true;
        }
        return q_.wait_dequeue_timed(out, std::chrono::milliseconds(timeout_ms));
    }

    /**
     * @brief 非阻塞出队（立即返回）
     */
    bool TryDequeue(std::shared_ptr<PacketBlob> &out)
    {
        return q_.try_dequeue(out);
    }

    /**
     * @brief （内部/外部都可用）入队：达到软上限时，先丢弃一个最旧元素，再插入新元素
     */
    void Enqueue(std::shared_ptr<PacketBlob> blob)
    {
        if (!blob)
            return;

        // 如果队列长度已到软上限，先丢弃一个最旧元素（追最新，减少堆积/延迟）
        if (q_.size_approx() >= capacity_soft_)
        {
            std::shared_ptr<PacketBlob> dropped;
            if (q_.try_dequeue(dropped))
            {
                ++stats_discarded_;
            }
        }

        q_.enqueue(std::move(blob));
        ++stats_enqueued_;
    }

    /**
     * @brief 清空队列（Stop(drain=false) 时会调用）
     */
    void DrainQueue()
    {
        std::shared_ptr<PacketBlob> tmp;
        while (q_.try_dequeue(tmp))
        {
            /* 释放即丢 */
        }
    }

    /**
     * @brief 统计：已成功入队的包数量
     */
    uint64_t EnqueuedCount() const { return stats_enqueued_.load(); }

    /**
     * @brief 统计：因软上限触发而丢弃的包数量
     */
    uint64_t DiscardedCount() const { return stats_discarded_.load(); }

private:
    //==================== 构造/析构与禁止拷贝 ====================

    /**
     * @brief 构造：设定默认容量与计数初值；原子布尔初值为 false
     */
    StreamBus()
        : capacity_soft_(300) // 默认软上限：300 个包（可按你的吞吐/延迟调参）
          ,
          stats_enqueued_(0), stats_discarded_(0), opened_(false), running_(false)
    {
    }

    /**
     * @brief 析构：确保线程退出并释放资源
     */
    ~StreamBus()
    {
        Stop(/*drain=*/true);
        Close();
    }

    StreamBus(const StreamBus &) = delete;
    StreamBus &operator=(const StreamBus &) = delete;

private:
    //==================== 成员变量 ====================

    // 1) 包队列：**存的是指针**（shared_ptr<PacketBlob>），避免在队列中复制大块二进制数据
    moodycamel::BlockingConcurrentQueue<std::shared_ptr<PacketBlob>> q_;

    // 2) 软上限容量与统计（注意：统计用于监控，不需强一致）
    size_t capacity_soft_;                  ///< 队列软上限：达到时入队会先丢弃一个最旧元素
    std::atomic<uint64_t> stats_enqueued_;  ///< 计数：入队次数
    std::atomic<uint64_t> stats_discarded_; ///< 计数：因软上限丢弃次数

    // 3) 拉流器与线程
    std::unique_ptr<VideoPacketReader> reader_; ///< 拉流器（封装 FFmpeg 打开/读取/回调）
    std::thread worker_;                        ///< 后台拉流线程

    // 4) 控制与状态
    std::atomic<bool> stop_{false};    ///< 停止标志（通知读循环退出）
    std::atomic<int64_t> seq_{0};      ///< 包序号（回调里原子自增）
    std::string source_id_;            ///< 当前源标识（写入 JSON，便于下游区分）
    std::atomic<bool> opened_{false};  ///< 【原子】是否已成功打开
    std::atomic<bool> running_{false}; ///< 【原子】拉流线程是否在运行
};

// =====================================================
// =============== 序列化：JSON + 二进制 ===============
// =====================================================

/**
 * @brief 将一帧（PacketBlob）序列化为**连续内存**：
 *        Header(4B 小端 JSON长度) + JSON(元数据) + extradata + data
 *
 * @param b  输入帧（只读）
 * @return   std::shared_ptr<std::vector<char>> 指向连续字节缓冲，可直接 write()/send()
 *
 * 设计说明：
 *   - JSON：只放“人/机可读的元数据”（不放二进制），便于抓包/调试与下游快速解析；
 *   - extradata：放在 JSON 之后、data 之前，接收端拿到 JSON 后可立即按长度读出 extradata，
 *                先初始化/更新解码器，再紧接着投喂 data；
 *   - data：压缩比特流（通常已为 Annex-B），长度与内容按原样拷贝；
 *   - 本函数做**严格的尺寸溢出检查**，避免极端情况下 size_t 溢出。
 */
inline std::shared_ptr<std::vector<char>> BuildFrameChunkBuffer(const PacketBlob &b)
{
    using nlohmann::json;

    // ---------- 1) 规范化长度（负值按 0 处理；转为 size_t） ----------
    const size_t data_sz = (b.data_size > 0) ? static_cast<size_t>(b.data_size) : size_t{0};
    const size_t extra_sz = (b.extradata_size > 0) ? static_cast<size_t>(b.extradata_size) : size_t{0};

    // ---------- 2) 组装 JSON 元数据 ----------
    json j;

    // 基本识别
    j["type"] = 106;
    j["source_id"] = b.source_id; // 源标识（如摄像头ID）
    j["seq"] = b.seq;             // 帧序列号

    // 编解码参数
    j["codec_id"] = b.codec_id;          // AVCodecID（整数）
    j["codec"] = b.codec_name;           // 短名（如 "h264"）
    j["codec_long"] = b.codec_long_name; // 长名（可读性更好）
    j["width"] = b.width;
    j["height"] = b.height;
    j["fps"] = b.fps;
    j["bit_rate"] = b.bit_rate;
    j["pix_fmt"] = b.pix_fmt_name;

    // 时间基与时间戳（接收端用：秒 = pts * num / den）
    j["time_base"] = {{"num", b.time_base_num}, {"den", b.time_base_den}};
    j["pts"] = b.pts;
    j["dts"] = b.dts;
    j["duration"] = b.duration;
    j["key"] = b.is_keyframe;

    // 长度信息：让接收端按长度读取二进制段
    j["extradata_size"] = static_cast<uint32_t>(extra_sz);
    j["data_size"] = static_cast<uint32_t>(data_sz);

    // JSON 串（UTF-8），不压缩，便于抓包与调试
    const std::string json_str = j.dump();

    // Header：4 字节小端存 JSON 长度（uint32）
    const uint32_t json_len_le = static_cast<uint32_t>(json_str.size());

    // ---------- 3) 计算总长度并做溢出保护 ----------
    auto add_or_throw = [](size_t a, size_t b) -> size_t
    {
        if (a > (std::numeric_limits<size_t>::max)() - b)
        {
            throw std::overflow_error("BuildFrameChunkBuffer: total size overflow");
        }
        return a + b;
    };

    size_t total = 0;
    total = add_or_throw(total, size_t{4});       // Header: 4B JSON长度
    total = add_or_throw(total, json_str.size()); // JSON
    total = add_or_throw(total, extra_sz);        // extradata
    total = add_or_throw(total, data_sz);         // data

    // ---------- 4) 分配连续缓冲并按顺序拷贝 ----------
    auto buf = std::make_shared<std::vector<char>>();
    buf->resize(total);

    char *dst = buf->data();
    size_t off = 0;

    // 4.1 写 Header（4 字节小端 JSON长度）
    dst[off + 0] = static_cast<char>(json_len_le & 0xFF);
    dst[off + 1] = static_cast<char>((json_len_le >> 8) & 0xFF);
    dst[off + 2] = static_cast<char>((json_len_le >> 16) & 0xFF);
    dst[off + 3] = static_cast<char>((json_len_le >> 24) & 0xFF);
    off += 4;

    // 4.2 写 JSON 文本
    if (!json_str.empty())
    {
        std::memcpy(dst + off, json_str.data(), json_str.size());
        off += json_str.size();
    }

    // 4.3 写 extradata（若存在）
    if (extra_sz > 0 && b.extradata_ptr)
    {
        std::memcpy(dst + off, b.extradata_ptr.get(), extra_sz);
        off += extra_sz;
    }

    // 4.4 写 data（若存在）
    if (data_sz > 0 && b.data_ptr)
    {
        std::memcpy(dst + off, b.data_ptr.get(), data_sz);
        off += data_sz;
    }

    // （可选）调试断言：确保偏移与 total 一致
    // assert(off == total);

    return buf;
}

/**
 * @brief 便捷函数：将一帧写入到文件（Header + JSON + extradata + data）
 * @return true 写入成功；false 写入失败（包括 fwrite 不完整）
 *
 * 说明：实际网络发送场景下，建议直接拿 BuildFrameChunkBuffer 的返回值走 sendall。
 */
inline bool WriteFrameChunk(FILE *f, const PacketBlob &b)
{
    auto buf = BuildFrameChunkBuffer(b);
    if (!buf || buf->empty())
        return false;
    const size_t n = std::fwrite(buf->data(), 1, buf->size(), f);
    return n == buf->size();
}

#endif // STREAM_BUS_HPP
